package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.SeckillVoucherMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.service.IVoucherOrderService;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import io.reactivex.Single;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;


/**
 * <p>
 * 秒杀优惠券表，与优惠券是一对一关系 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2022-01-04
 */
@Service
public class SeckillVoucherServiceImpl extends ServiceImpl<SeckillVoucherMapper, SeckillVoucher> implements ISeckillVoucherService {
    @Resource
    private RedisIdWorker  redisIdWorker;
    @Resource
    private IVoucherOrderService  voucherOrderService;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;
    private static final DefaultRedisScript<Long> redisScript ;
    static {
        redisScript = new DefaultRedisScript<>();
        redisScript.setLocation(new ClassPathResource("redis.lua"));
        redisScript.setResultType(Long.class);
    }
    private BlockingQueue<VoucherOrder>  voucherTask= new ArrayBlockingQueue<>(1024*1024);
    //一个线程的线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    //线程任务,为了提高性能，肯定在秒杀订单前开启，Spring提供一个注解可以在初始化类后开启
    @PostConstruct
    private void init(){
        //初始化之后进行调用内部类
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    //        * 秒杀是否开始或结束，如果尚未开始或已经结束则无法下单
//        * 库存是否充足，不足则无法下单
    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        //获取用户id
        Long userId = UserHolder.getUser().getId();
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        //1、执行lua脚本
       Long success =  stringRedisTemplate.execute(
                redisScript,
                Collections.emptyList(),
                voucherId.toString(),
                userId.toString(),
               String.valueOf(orderId)
       );
        //2、判断结果是否为0
        int r = success.intValue();
        System.out.println(r);
        if(r!=0){
            //否返回异常信息
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        //2.1是，将优惠卷id、用户id合订单id存入阻塞队列中
        //封装订单，放置阻塞队列
        voucherOrder.setId(orderId);
        //返回订单id
        //用户id
        voucherOrder.setUserId(userId);
        //代金卷的id
        voucherOrder.setVoucherId(voucherId);
        //TODO 保存阻塞队列
        voucherTask.add(voucherOrder);

        //完成了抢单情况
        //开启新的线程进行异步下单
//        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
        //返回订单id
        return Result.ok(orderId);
    }


//    private class VoucherOrderHandler implements Runnable {
//        @Override
//        public void run() {
//            //持续获取队列订单信息
//            //会持续循环检测阻塞队列，阻塞队列如果没有订单信息会退出
//            while (true) {
//                try {
//                    //1、获取队列中的订单信息 XREADGROUP Group g1 c1 count 1 block 2000 Streams streams.order
//                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
//                            Consumer.from("g1", "c1"),
//                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
//                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
//                    );
//                    //2、判断消息获取是否成功
//                    if(list==null||list.isEmpty()){
//                        //2.1如果获取成功，说明没有消息，继续下一次循环
//                        continue;
//                    }
//                    //解析消息中的订单消息
//                    MapRecord<String,Object,Object> record = list.get(0);
//                    Map<Object,Object> values = record.getValue();
//                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values,new VoucherOrder(),true);
//                    //3如果获取成功，说明可以下单
//                    handleVoucherOrder(voucherOrder);
//                    //ACK确认
//                    stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());
//                }catch (Exception e){
//                    log.error("处理订单异常",e);
//                    handlePendingList();
//                }
//            }
//        }



    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            //持续获取队列订单信息
            //会持续循环检测阻塞队列，阻塞队列如果没有订单信息会退出
            while (true) {
                try {
                    //1、获取队列中的订单信息
                    VoucherOrder voucherOrder = voucherTask.take();
                    //2、创建订单
                    handleVoucherOrder(voucherOrder);
                    //创建锁
                }catch (Exception e){
                    log.error("处理订单异常",e);
                }
            }
        }

        private void handleVoucherOrder(VoucherOrder voucherOrder) {
            Long UserId = voucherOrder.getUserId();
            RLock lock = redissonClient.getLock("lock:order:" + UserId);
            if(!lock.tryLock()) {
                // 获取锁失败，直接返回失败或者重试
                log.error("不允许重复下单！");
                return;
            }
            //获取锁对象
            try {
//                IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
                voucherOrderService.createVoucherOrder(voucherOrder);
            } catch (IllegalStateException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }

        }
//        private void handlePendingList() {
//            while (true) {
//                try {
//                    //1、获取队列中的订单信息 XREADGROUP Group g1 c1 count 1 block 2000 Streams streams.order
//                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
//                            Consumer.from("g1", "c1"),
//                            StreamReadOptions.empty().count(1),
//                            StreamOffset.create("stream.orders", ReadOffset.from("0"))
//                    );
//                    //2、判断消息获取是否成功
//                 if (list == null || list.isEmpty()) {
//                        //2.1如果获取成功，说明没有消息，继续下一次循环
//                        break;
//                    }
//                    //解析消息中的订单消息
//                    MapRecord<String, Object, Object> record = list.get(0);
//                    Map<Object, Object> values = record.getValue();
//                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
//                    //3如果获取成功，说明可以下单
//                    handleVoucherOrder(voucherOrder);
//                    //ACK确认
//                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
//                } catch (Exception e) {
//                    log.error("处理订单异常", e);
//                    try {
//                        Thread.sleep(20);
//                    } catch (InterruptedException ex) {
//                        throw new RuntimeException(ex);
//                    }
//                }
//            }
//
//        }
    }
}

//    @Override
//    @Transactional
//    public Result seckillVoucher(Long voucherId) {
//        //1、查询优惠卷信息
//        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
//        System.out.println(seckillVoucher.toString());
//        //获取信息
//        LocalDateTime beginTime = seckillVoucher.getBeginTime();
//        LocalDateTime endTime = seckillVoucher.getEndTime();
//        //2、判断秒杀是否已经开始
//        if(LocalDate   Time.now().isBefore(beginTime) ||LocalDateTime.now().isAfter(endTime)){
//            //3、否，直接返回异常，结束
//            return Result.fail("秒杀卷还未开始或结束");
//        }
//        //是，判断库存是否充足
//        if(seckillVoucher.getStock()<=0)
//        {
//            //否，直接返回异常
//            return Result.fail("卷已被抢完");
//        }
////        //是扣减库存
////        boolean success = seckillVoucherService.update().setSql("stock = stock -1")
////                .eq("voucher_Id", voucherId).gt("stock", 0)
////                .update();
////        if(!success){
////            return Result.fail("库存不足!");
////        }
////        //创建订单
////        VoucherOrder voucherOrder = new VoucherOrder();
////        long orderId = redisIdWorker.nextId("order");
////        voucherOrder.setId(orderId);
////
////        //返回订单id
////        //用户id
////        Long userId = UserHolder.getUser().getId();
////        voucherOrder.setUserId(userId);
////        //代金卷的id
////        voucherOrder.setVoucherId(voucherId);
////        //添加订单
//
//        Long UserId = UserHolder.getUser().getId();
//        //创建锁
////        SimpleredisLock simpleredisLock = new SimpleredisLock(stringRedisTemplate,"order"+UserId);
//        //使用锁
////        boolean tryLock = simpleredisLock.tryLock(1200);
//        RLock lock = redissonClient.getLock("lock:order:" + UserId);
//        //获取锁对象
//        boolean tryLock = lock.tryLock();
//        if(!tryLock){
//            return Result.fail("不允许重复下单");
//        }
///*
//    分布式锁，利用setnx创建锁，用户需要先进行获取，如：单个对象的并发进程，当一个进程占用后，进程创建对象，另一个进程创建时，无法创建成功，返回
//    当遇到集群现象时，和前面情况类似，相比synchronized，synchronized对于锁的对象 面对集群是无法成功的，synchronized只能锁住一个jvm对象
// */
////        synchronized(UserId.toString().intern()){
//            //获取代理对象
//        try {
//            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
//            return proxy.createVoucherOrder(voucherId);
//        } catch (IllegalStateException e) {
//            throw new RuntimeException(e);
//        } finally {
//            lock.unlock();
//        }
//        }

